package com.ibm.factories;

import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;

import com.google.common.base.Throwables;
import com.ibm.cps.message.MessageFactory;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.RawDataMessage;
import com.ibm.cps.newmessage.BasicDataSourceMetadata;
import com.ibm.interfaces.IMessageParser;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.exception.CPSException;

public class StaticDataSourceFactory {


	private static final Logger logger = Logger.getLogger(DataSourceFactory.class);

	private Map<String, BasicDataSourceMetadata> name2DataSourceMap;

	public StaticDataSourceFactory() {
		name2DataSourceMap = new HashMap<String, BasicDataSourceMetadata>();
	}

	public void addDataSource(BasicDataSourceMetadata metadata) {
		if (metadata == null) {
			return;
		}
		String dsname = metadata.getName();
		if (name2DataSourceMap.get(dsname) != null) {
			logger.warn("Data source " + dsname + " was already existed.");
			return;
		}
		name2DataSourceMap.put(metadata.getName(), metadata);
	}

	public void deleteDataSource(BasicDataSourceMetadata metadata) {
		if (metadata == null) {
			return;
		}
		String dsname = metadata.getName();
		name2DataSourceMap.remove(dsname);
	}

	public ParsedDataMessage parseRawData(RawDataMessage rawDataMessage)
			throws CPSException {
		String sourceid = rawDataMessage.getSourceid();
		if (name2DataSourceMap.get(sourceid) == null) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
					"No data source metadata for data source " + sourceid + ".");
		}
		try {
			ITimeSeriesSchema schema = name2DataSourceMap.get(sourceid)
					.getTimeSeriesSchema();
			IMessageParser schemaParser = name2DataSourceMap.get(sourceid).getTimeSeriesParser(schema);
			ParsedDataMessage parsedMsg = MessageFactory
					.produceParsedDataMessage(schemaParser, rawDataMessage);
			return parsedMsg;
		} catch (Exception e) {
			logger.error(Throwables.getStackTraceAsString(e));
		}
		return null;
	}
	

	/**
	 * get input processor metadata according to data source name
	 * 
	 * @param topicName
	 * @return
	 * @throws CPSException
	 */
	public String replicateDataSource(String dsname) throws CPSException {
		if (dsname == null || name2DataSourceMap.get(dsname) == null) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
					"There is no metadata for data soruce " + dsname + ".");
		}
		return name2DataSourceMap.get(dsname).replicate();
	}

	public boolean containsDataSource(String dataSourceId) {
		return name2DataSourceMap.containsKey(dataSourceId);
	}

	public BasicDataSourceMetadata getDataSourceMetadata(String dsname) {
		if (containsDataSource(dsname)) {
			return name2DataSourceMap.get(dsname);
		} else {
			return null;
		}
	}

}
